/*
 * Copyright 2009 by Konstantin Bulenkov
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.bulenkov.crawler.model;

import com.bulenkov.crawler.CURL;
import com.bulenkov.crawler.CrawlerThreadResult;
import com.bulenkov.crawler.event.CrawlerListener;
import com.bulenkov.crawler.filter.Filter;
import com.bulenkov.crawler.filter.FilterFactory;

import java.io.*;
import java.net.MalformedURLException;
import java.util.*;

/**
 * @author Konstantin Bulenkov
 */
public class SimpleCrawlerModel implements StatefullCrawlerModel, Serializable {
  private final List<CURL> links = new ArrayList<CURL>();
  private transient final Map<IteratorOwnerInfo, CURL> activeThreads = new HashMap<IteratorOwnerInfo, CURL>();
  private final List<CrawlerListener> listeners = new ArrayList<CrawlerListener>();  

  private volatile int seek = 0;
  private Filter filter = FilterFactory.ALLOW_ALL;
  private Filter stopFilter = FilterFactory.DISALLOW_ALL;

  public SimpleCrawlerModel(String... startLinks) {
    for (String link : startLinks) {
      try {
        links.add(new CURL(link));
      } catch (MalformedURLException e) {//
        //TODO: add a log message
      }
    }
  }

  public List<CURL> getLinksFound() {
    return Collections.unmodifiableList(links);
  }

  public void remove(CURL link) {
    throw new UnsupportedOperationException();
  }

  public void addListener(CrawlerListener listener) {
    listeners.add(listener);
  }

  public void removeListener(CrawlerListener listener) {
    listeners.remove(listener);
  }

  public long getCurrentIndex() {
    return seek;
  }

  public void setStopFilter(Filter filter) {
    stopFilter = filter == null ? FilterFactory.DISALLOW_ALL : filter;
  }

  public void markUnreachable(CURL link) {
    //TODO: make it works
  }

  public synchronized boolean hasNext() {
    if (seek < links.size()) {
      activeThreads.put(new IteratorOwnerInfo(), links.get(seek));
      seek++;
      return true;
    }

    while (isWaitingForNextURLs() || hasNextStopURL()) {
      if (hasNextStopURL()) {
        seek++;
        continue;
      }

      try {        
        wait(); //waiting for notify from method add(CrawlerThreadResult)
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }

    if (seek < links.size()) {
      activeThreads.put(new IteratorOwnerInfo(), links.get(seek));
      seek++;
      return true;
    }
    return false;
  }

  private boolean hasNextStopURL() {
    return seek < links.size() && stopFilter.accept(links.get(seek));
  }

  public CURL next() {
    final CURL curl = activeThreads.get(new IteratorOwnerInfo());
    if (curl == null) {
      throw new ConcurrentModificationException("WTF");
    }
    return curl;
  }

  private long start = System.currentTimeMillis();

  public synchronized void add(CrawlerThreadResult result) {
    final List<CURL> added = new ArrayList<CURL>();
    for (CURL link : result.getFoundURLs()) {
      if (! links.contains(link) && isAccepted(link)) {
        added.add(link);
      }
    }
    links.addAll(added);
    activeThreads.remove(new IteratorOwnerInfo());
    notifyAll();
    notifyListeners(added);
  }

  private void notifyListeners(List<CURL> added) {
    new Notifier(added.toArray(new CURL[added.size()])).start();
  }

  private boolean isAccepted(CURL link) {
    return (link != null && link.url() != null && (filter == null || filter.accept(link)));
  }

  public void setFilter(Filter filter) {
    this.filter = filter == null ? FilterFactory.ALLOW_ALL : filter;
  }

  public Filter getFilter() {
    return null;
  }

  public synchronized void remove() {
    throw new UnsupportedOperationException("Prohibited");
  }

  public synchronized void save(OutputStream out) throws UnsupportedOperationException {
    try {
      new ObjectOutputStream(out).writeObject(this);
    } catch (IOException e) {
      e.printStackTrace();
    }
  }

  public synchronized void load(InputStream in) throws UnsupportedOperationException {
    try {
      Object o = new ObjectInputStream(in).readObject();
      if (o instanceof SimpleCrawlerModel) {
        final SimpleCrawlerModel m = (SimpleCrawlerModel) o;
        links.clear();
        links.addAll(m.links);
        seek = m.seek;
      }
    } catch (IOException e) {
      e.printStackTrace();
    } catch (ClassNotFoundException e) {
      e.printStackTrace();
    }
  }

  public boolean isWaitingForNextURLs() {
    return !activeThreads.isEmpty() && seek == links.size();
  }

  private static class IteratorOwnerInfo {
    private final Thread thread;
    private final long startTime;

    public IteratorOwnerInfo() {
      thread = Thread.currentThread();
      startTime = System.currentTimeMillis();
    }

    @Override
    public boolean equals(Object o) {
      if (this == o) return true;
      if (o == null || getClass() != o.getClass()) return false;

      final IteratorOwnerInfo that = (IteratorOwnerInfo) o;

      return thread.equals(that.thread);
    }

    @Override
    public int hashCode() {
      return thread.hashCode();
    }

    public long timeElapsed() {
      return System.currentTimeMillis() - startTime;
    }
  }

  class Notifier extends Thread {
    private CURL[] curls;


    public Notifier(CURL... url) {
      super("ListenersNotifier");
      curls = url;
    }

    @Override
    public void run() {
      for (CrawlerListener listener : listeners) {
        for (CURL curl : curls) {
          if (listener.getLinkFilter().accept(curl)) {
            listener.linkAdded(curl);
          }
        }
      }
    }
  }
}
